package cn.com.basic.framework.datahub.proxy;

import com.aliyun.datahub.client.DatahubClient;
import com.aliyun.datahub.client.exception.DatahubClientException;
import com.aliyun.datahub.client.exception.SubscriptionOffsetResetException;
import com.aliyun.datahub.client.model.*;
import com.aliyun.datahub.exception.OffsetSessionChangedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

/**
 * Datahub 读取类负责对单个通道进行订阅读取
 *
 * @author zhuxj
 */
public class ReadProxy {

    private static Logger logger = LoggerFactory.getLogger(ReadProxy.class);


    private String projectName;             //项目空间
    private String topicName;               //topic名称
    private String subId;                   //订阅号
    private DatahubClient client;

    private OpenSubscriptionSessionResult sessionResult;

    private String shardId;
    private List<String> shardIds = new ArrayList<>();

    private Consumer<TupleRecordData> consumer;

    /**
     * 唯一构造
     *
     * @param projectName 项目空间名称
     * @param topicName   主题名称
     * @param subId       订阅号
     */
    public ReadProxy(DatahubClient client, String projectName, String topicName, String subId) {
        this.projectName = projectName;
        this.topicName = topicName;
        this.subId = subId;
        this.client = client;
    }

    /**
     * 开始读取
     */
    public void startRead() {
        for (int i = 0; i < client.getTopic(this.projectName, this.topicName).getShardCount(); i++) {
            this.shardIds.add(i + "");
        }

        SubscriptionOffset subscriptionOffset = sessionResult.getOffsets().get(shardId);
        String cursor = "";
        try {
            if (subscriptionOffset.getSequence() >= 0) {
                // 获取下一条记录的Cursor
                long nextSequence = subscriptionOffset.getSequence();
                // 备注：如果按照SEQUENCE getCursor报SeekOutOfRange错误，需要回退到按照SYSTEM_TIME或者OLDEST/LATEST进行getCursor
                cursor = client.getCursor(projectName, topicName, shardId, CursorType.SEQUENCE, nextSequence).getCursor();
            } else {
                // 获取最旧数据的Cursor
                cursor = client.getCursor(projectName, topicName, shardId, CursorType.OLDEST).getCursor();
            }
        } catch (Exception e) {
            // 获取最旧数据的Cursor
            cursor = client.getCursor(projectName, topicName, shardId, CursorType.OLDEST).getCursor();
        }


        RecordSchema schema = client.getTopic(projectName, topicName).getRecordSchema();

        int fetchNum = 100;
        long recordCount = 0L;
        while (true) {
            try {
                GetRecordsResult recordsResult = client.getRecords(projectName, topicName, shardId, schema, cursor, fetchNum);
                if (recordsResult.getRecordCount() <= 0) {
                    try {
                        Thread.sleep(1000);
                        continue;
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                for (RecordEntry recordEntry : recordsResult.getRecords()) {

                    TupleRecordData recordData = (TupleRecordData) recordEntry.getRecordData();

                    consumer.accept(recordData);

                    // 处理数据完成后，设置点位
                    ++recordCount;
                    if (recordCount % 1000 == 0) {
                        subscriptionOffset.setSequence(recordEntry.getSequence());
                        subscriptionOffset.setTimestamp(recordEntry.getSystemTime());
                        Map<String, SubscriptionOffset> offsetMap = new HashMap<>();
                        offsetMap.put(shardId, subscriptionOffset);
                        client.commitSubscriptionOffset(projectName, topicName, subId, offsetMap);
                        recordCount = 0;
                    }
                }
                cursor = recordsResult.getNextCursor();
            } catch (SubscriptionOffsetResetException e) {
                logger.warn("点位发生重置,请确认是否属于正常操作!!!");
                // 表示点位被重置，重新获取SubscriptionOffset信息，这里以Sequence重置为例
                // 如果以Timestamp重置，需要通过CursorType.SYSTEM_TIME获取cursor
                //环境1
                subscriptionOffset = client.getSubscriptionOffset(projectName, topicName, subId, shardIds).getOffsets().get(shardId);
                long nextSequence = subscriptionOffset.getSequence() + 1;
                cursor = client.getCursor(projectName, topicName, shardId, CursorType.SEQUENCE, nextSequence).getCursor();

                //环境2
//                 cursor = client.getCursor(projectName, topicName, shardId, CursorType.SYSTEM_TIME).getCursor();
            } catch (OffsetSessionChangedException | DatahubClientException e) {
                // 退出. Offline: 订阅下线; SessionChange: 表示订阅被其他客户端同时消费
                logger.error(e.getMessage(), e);
            }
        }
    }


    public void setSessionResult(OpenSubscriptionSessionResult sessionResult) {
        this.sessionResult = sessionResult;
    }

    public void setShardId(String shardId) {
        this.shardId = shardId;
    }

    public void setConsumer(Consumer<TupleRecordData> consumer) {
        this.consumer = consumer;
    }
}
